import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object StateProgramingTest01 {
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream: DataStream[String] = env.socketTextStream("localhost", 7777)

    val tempDS = stream.map(
      data => {
        val dataArr = data.split(",")
        TemperatureData(dataArr(0), dataArr(1).toLong, dataArr(2).toDouble)
      }
    )

    val keyedData = tempDS.keyBy(_.id)

    keyedData.flatMapWithState(new TemperatureAlertFuntion(1.7))

  }
  case class TemperatureData(id: String, time: Long, temp: Double)

  /**
   * 通过 RuntimeContext 注册 StateDescriptor。
   * StateDescriptor 以状态 state 的名字和存储的数据类型为参数。
   * 在 open()方法中创建 state 变量。
   * @param threshold
   */
  class TemperatureAlertFuntion(threshold: Double) extends RichFlatMapFunction[TemperatureData, (String, Double, Double)] {

    private var lastTempState: ValueState[Double] = _

    override def open(parameters: Configuration): Unit = {
      val lastTempDescriptor = new ValueStateDescriptor[Double]("lastTemp", classOf[Double])
      lastTempState = getRuntimeContext.getState(lastTempDescriptor)
    }

    override def flatMap(reading: TemperatureData,
                         out: Collector[(String, Double, Double)]): Unit = {
      val lastTemp = lastTempState.value()
      val tempDiff = (reading.temp - lastTemp).abs
      if (tempDiff > threshold) {
        out.collect((reading.id, reading.temp, tempDiff))
      }
      this.lastTempState.update(reading.temp)
    }
  }
}
